package com.flute.haflute.agent.job;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.common.ClusterConstants.ROLE;

/**
 * 集群现有资源管理器<br>
 * 包括已经被分配的节点和空余节点的管理等
 */
public class JobResourceManager {
	
	private static Logger logger = LoggerFactory
			.getLogger(JobResourceManager.class);
	
	private static JobResourceManager instance = new JobResourceManager();

	/** busyNodes 已经分配执行工作的节点列表 */
	private Map<Node, JobContext> busyNodes;

	public static JobResourceManager getInstance() {
		return instance;
	}

	private JobResourceManager() {
		busyNodes = new ConcurrentHashMap<Node, JobContext>();
	}

	/**
	 * 返回所有不在busy列表中的从节点
	 */
	public List<Node> getFreeSlaves() {
		List<Node> result = new ArrayList<Node>();
		Map<Node, ROLE> nodes = ClusterTopoDBService.getInstance()
				.getNodeRoleDB();
		for (Entry<Node, ROLE> entry : nodes.entrySet()) {
			if (!busyNodes.containsKey(entry.getKey())
					&& entry.getValue() == ROLE.SLAVE) {
				result.add(entry.getKey());
			}
		}
		return result;
	}

	/**
	 * 返回所有不在busy列表中的节点，包括BDM，DM等
	 */
	public synchronized List<Node> getAllFreeNodes() {
		List<Node> result = new ArrayList<Node>();
		Map<Node, ROLE> nodes = ClusterTopoDBService.getInstance()
				.getNodeRoleDB();
		for (Entry<Node, ROLE> entry : nodes.entrySet()) {
			if (!busyNodes.containsKey(entry.getKey())) {
				result.add(entry.getKey());
			}
		}
		return result;
	}

	/**
	 * 注册已经被预分配的节点，该方法被设计为只被任务调度前的上下文资源申请时使用。
	 * 
	 * @param newBusyNodes
	 */
	public synchronized void registerAllocatedNodes(
			Map<Node, JobContext> newBusyNodes) {
		busyNodes.putAll(newBusyNodes);
	}

	/**
	 * 查询某节点是否已经被分配
	 * 
	 * @param node
	 *            节点
	 * @return boolean
	 */
	public boolean isNodeBusy(Node node) {
		return busyNodes.get(node) != null;
	}

	/**
	 * 释放已经完成某项工作的节点
	 * 
	 * @param newFreeNodes
	 *            新的空余节点（即已经完成工作还没有分配新工作的节点）
	 * 
	 */
	public synchronized void releaseAllocatedNodes(
			Map<Node, JobContext> newFreeNodes) {
		if (newFreeNodes == null || newFreeNodes.isEmpty())
			return;

		for (Iterator<Node> ite = newFreeNodes.keySet().iterator(); ite
				.hasNext();) {
			busyNodes.remove(ite.next());
		}
	}

	/**
	 * @param jobController
	 * @return 释放的非Controller（Job Master）节点，即不包括@param jobController
	 */
	public synchronized List<Node> releaseNodes4Job(Node jobController) {
		List<Node> releasedNodes = new ArrayList<Node>();
		JobContext context = busyNodes.get(jobController);
		if (context == null) {
			logger.info("not knowing busyNodes: {}", jobController);
			return releasedNodes;
		}
		context.decJobCount();
		if (context.getNumOfJob() > 0) {
			return releasedNodes;
		}

		logger.info("to release one Controll Node : {}", jobController);
		for (Node node : context.getJobSlaves()) {
			busyNodes.remove(node);
			releasedNodes.add(node);
			logger.info("release one Node: {}", node);
		}

		return releasedNodes;
	}

	public synchronized List<JobContext> getBusyJobContext() {
		List<JobContext> nodes = new ArrayList<JobContext>();
		nodes.addAll(busyNodes.values());
		return nodes;
	}

	public synchronized JobContext getBusyJobContext4Node(Node jobController) {
		return busyNodes.get(jobController);
	}
}
